﻿using org.apache.zookeeper;
using org.apache.zookeeper.data;
using Serilog;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using static org.apache.zookeeper.Watcher;

namespace RedisDemo
{

    public class ZooKeeperHelper : Watcher, IDisposable
    {
        private ZooKeeper _zooKeeper;
        private bool _isDisposed = false;
        private Event.KeeperState _currentState = Event.KeeperState.Disconnected;
        private AutoResetEvent _stateChangedEvent = new AutoResetEvent(false);
        private AutoResetEvent _pathChangeEvent = new AutoResetEvent(false);


        private string _connectionString;
        private TimeSpan _sessionTimeout;
        private long _sessionId;

        public ZooKeeperHelper(string connectionString, TimeSpan sessionTimeout, long sessionId)
        {
            this._connectionString = connectionString;
            this._sessionTimeout = sessionTimeout;
            this._sessionId = sessionId;
            this._zooKeeper = CreateZooKeeper(connectionString, sessionTimeout, sessionId);
        }

        private string _currentWatchPath;
        private bool _currentWatchPathDeleted = false;

        public async Task<bool> WatchPathDeleted(string path, TimeSpan timeout)
        {
            _currentWatchPath = path;
            await _zooKeeper.existsAsync(path, true);
            Log.Debug("WatchPathDelete: {path}", path);
            var waiting = true;
            while (!_currentWatchPathDeleted)
            {
                waiting = _pathChangeEvent.WaitOne(timeout);
                Log.Debug("WatchPathDelete: {path} waiting {waiting}", path, waiting);
                if (!waiting)
                {
                    return false;
                }
            }
            return true;
            //return _pathChangeEvent.WaitOne(timeout);

        }

        public bool WaitFor(Event.KeeperState keeperState, TimeSpan timeout)
        {
            while (this._currentState != keeperState)
            {
                //Log.Debug("Wait for {@state}", keeperState);
                var waitResult = this._stateChangedEvent.WaitOne(timeout);
                //Log.Debug("Wait for result {0}", waitResult);
                return waitResult;

            }
            return true;
        }

        public void Dispose()
        {
            if (this._isDisposed)
            {
                return;
            }
            _isDisposed = true;

            if (_zooKeeper != null)
            {
                _zooKeeper.closeAsync().ConfigureAwait(false).GetAwaiter().GetResult();
            }
        }

        public async Task<ChildrenResult> GetChildrenAsync(string path, bool watch)
        {
            try
            {
                return await this._zooKeeper.getChildrenAsync(path, watch);
            }
            catch
            {
                throw;
            }
        }

        public List<Watcher> watchers = new List<Watcher>();

        public void OnWatch(WatchedEvent we)
        {
            Log.Debug("OnWatch {0} {1} {2}", we.getPath(), we.get_Type(), we.getState());
            this.WatchPath(_currentParentPath, _currentId).ConfigureAwait(true).GetAwaiter()
                .GetResult();
        }

        string _currentParentPath;
        long _currentId;

        private EpemeralNodeWatcher watcher;
        public void InitWatcher()
        {
            this.watcher = new EpemeralNodeWatcher();
            this.watcher.OnProcess += this.OnWatch;
        }

        public async Task WatchPath(string parentPath, long currentId)
        {
            this._currentParentPath = parentPath;
            //this._currentNodePath = nodePath;
            this._currentId = currentId;



            //var watcher = watchers[0];
            //var path = parentPath + "/" + nodePath;
            Log.Debug("WatchPath Path {@path}, currentId: {currentId}", parentPath, currentId);

            //var watcher = new EpemeralNodeWatcher();
            //watcher.OnProcess += this.OnWatch;

            var childrenResult = await this.GetChildrenAsync("/zk", false);

            Log.Debug("Result {@r}", childrenResult.Children.Count);

            List<EpemeralNode> nodes = new List<EpemeralNode>();
            foreach (var c in childrenResult.Children)
            {
                var epemeralNode = new EpemeralNode();

                var reg = Regex.Match(c, "(\\d+)");
                if (reg.Success)
                {
                    epemeralNode.Id = long.Parse(reg.Groups[0].Value);
                }

                epemeralNode.Path = parentPath + "/" + c;
                nodes.Add(epemeralNode);
            }

            if (nodes.Select(node => node.Id).Min() == currentId)
            {
                _currentId = currentId - 1;

                Log.Debug($"task {currentId} acquire lock success");
                //var random = new Random();
                //var sleepSeconds = random.Next(10);
                //Thread.Sleep(sleepSeconds * 1000);
                this.Dispose();
            }
            else
            {
                Log.Debug("CurrentId {@id} wait lock release...", currentId);
                //this.watchers.Enqueue(watcher);

                //var prevNodeId = currentId - 1;
                //var prevPath = nodes.First(node => node.Id == prevNodeId);
                //Log.Debug($"task {it} wait lock release.... prevPath {prevPath.Path}");

                //var acqureLock = await zooKeeperHelper.WatchPathDeleted(prevPath.Path, TimeSpan.FromSeconds(10));
                //if (acqureLock)
                //{
                //    Log.Debug($"task {it} retry get lock success.");
                //    zooKeeperHelper.Dispose();
                //}
                //else
                //{
                //    Console.ReadLine();
                //}
                //Console.ReadLine();
            }
        }

        public async Task<ChildrenResult> GetChildrenAsync(string path, Watcher watcher)
        {
            try
            {
                return await this._zooKeeper.getChildrenAsync(path, watcher);
            }
            catch
            {
                throw;
            }
        }

        public async Task<string> CreateAsync(string path, string data, List<ACL> acls, CreateMode createMode)
        {
            var result = await this._zooKeeper.createAsync(path, Encoding.UTF8.GetBytes(data), acls, createMode);
            return result;
        }

        public override Task process(WatchedEvent @event)
        {
            Log.Debug("Watcher#process type: {0}, state: {1}, path: {2}", @event.get_Type(), @event.getState(), @event.getPath());
            //Console.WriteLine("[{1}]Watcher#@event type: {0}, state: {2}, path: {3}", @event.get_Type(), DateTime.Now,
            //    @event.getState(), @event.getPath());
            this._currentState = @event.getState();

            if (@event.getState() == Event.KeeperState.Expired)
            {
                if (_zooKeeper != null)
                {
                    try
                    {
                        _zooKeeper.closeAsync().ConfigureAwait(false).GetAwaiter().GetResult();
                    }
                    catch
                    {
                        // ignored
                    }
                }

                this._zooKeeper = CreateZooKeeper(this._connectionString, this._sessionTimeout, this._sessionId);
            }
            else if (@event.getState() == Event.KeeperState.SyncConnected)
            {
                if (!string.IsNullOrEmpty(_currentWatchPath))
                {
                    var path = @event.getPath();
                    if (path.Equals(_currentWatchPath, StringComparison.OrdinalIgnoreCase))
                    {
                        _currentWatchPathDeleted = true;
                        _pathChangeEvent.Set();
                    }
                }

                //this.GetChildren(@event.getPath()).ConfigureAwait(true).GetAwaiter();
                //this._zooKeeper.
            }

            //Console.WriteLine($"[{DateTime.Now}] event set true");
            //Log.Debug("_stateChangeEvent.Set()");


            this._stateChangedEvent.Set();

            return Task.CompletedTask;
        }

        private ZooKeeper CreateZooKeeper(String connectionString, TimeSpan sessionTimeout, long sessionId)
        {
            return new ZooKeeper(connectionString, (int)sessionTimeout.TotalMilliseconds, this);
        }
    }

    public class EpemeralNode
    {
        public long Id { get; set; }
        public string Path { get; set; }
    }

    public class EpemeralNodeWatcher : Watcher
    {
        public event Action<WatchedEvent> OnProcess;

        public override Task process(WatchedEvent @event)
        {
            if (OnProcess != null)
            {
                OnProcess(@event);
            }
            return Task.CompletedTask;
        }
    }

    public class MyWatch : Watcher
    {

        public string WatchedPath { get; set; }
        public ZookeeperDemo ZkDemo { get; set; }

        public string EpemeralPath { get; set; }

        public long NodeId { get; set; }

        public Task ViewChildren()
        {
            ////ZkDemo.WatchPath(this.WatchedPath);
            var task = Task.Run(() =>
            {
                //ZkDemo.FireAcquireLock();
              
                var reuslt = ZkDemo.ZK.getChildrenAsync(WatchedPath, this).Result;

                var id = ZkDemo.GetIdFromPath(this.EpemeralPath);

                


                //Log.Debug("WatchPath Result {@r}", reuslt.Children);

                var ids = new List<long>();
                foreach (var item in reuslt.Children)
                {
                    ids.Add(ZkDemo.GetIdFromPath(item));
                    //Log.Debug("Children: {@c}", item);
                }

                if (ids.Count > 0 && ids.Min() == id)
                {
                    lock (ZkDemo)
                    {
                        Log.Debug("CurrentId {@id} Check itself is min Id", id);
                        ZkDemo.HasAcquireLock = true;
                        ZkDemo.FireAcquireLock(id);
                    }
                }
            });
            return task;
        }

        public override Task process(WatchedEvent @event)
        {
            if (ZkDemo.HasAcquireLock)
            {
                Log.Debug("NodeId {@id} Has Acquire Lock return", this.NodeId);
                return Task.CompletedTask;
            }

            //Log.Debug("path: {@path}, state: {@state}, type: {@type}", @event.getPath(), @event.getState(), @event.get_Type());

            ////ZookeeperDemo.TestWatch(this).ConfigureAwait(false);
            //Log.Debug("Callback ZkDemo#WatchePath {@path} {@tid}", this.WatchedPath, Thread.CurrentThread.ManagedThreadId);

            Log.Debug("Start NodeId {@id} Watcher Callback", this.NodeId);
            ViewChildren();
            ZkDemo.ZK.getChildrenAsync(WatchedPath, this);
            //Log.Debug("End NodeId {@id} Watcher Callback", this.NodeId);
            ////ZkDemo.WatchPath(this.WatchedPath);
            //var reuslt = ZkDemo.ZK.getChildrenAsync(WatchedPath, this).Result;

            //Log.Debug("WatchPath Result {@r}", reuslt.Children);

            //foreach (var item in reuslt.Children)
            //{
            //    Log.Debug("Children: {@c}", item);
            //}


            //Log.Debug("End Callback ZkDemo#WatchePath {@path} {@tid}", this.WatchedPath, Thread.CurrentThread.ManagedThreadId);

            return Task.CompletedTask;
        }
    }

    public class ZkInitWatcher : Watcher
    {
        public AutoResetEvent EventCondition { get; set; }
        public event Action<WatchedEvent> Callback;

        public override Task process(WatchedEvent @event)
        {
            if (@event.getState() == Event.KeeperState.SyncConnected)
            {
                Callback(@event);
                if (EventCondition != null)
                {
                    EventCondition.Set();
                }
            }
            return Task.CompletedTask;
        }
    }

    public class ZookeeperDemo
    {

        public ZooKeeper ZK = null;
        private Event.KeeperState state;
        private AutoResetEvent notifyEvent = new AutoResetEvent(false);
        public event Func<long, Task> OnAcquireLock;

        public Task FireAcquireLock(long id)
        {
            //Log.Debug("NodeId {@id} FireAcquireLock Event....", id);
            this.OnAcquireLock(id).Wait();
            //Log.Debug("NodeId {@id} Close Zk....", id);
            this.ZK.closeAsync().ConfigureAwait(false).GetAwaiter().GetResult();
            Log.Debug("NodeId {@id} Close Zk Success", id);
            return Task.CompletedTask;
        }

        public void init()
        {
            var zkInitWatcher = new ZkInitWatcher();
            zkInitWatcher.Callback += (we) =>
            {
                this.state = we.getState();
            };
            zkInitWatcher.EventCondition = notifyEvent;

            ZK = new ZooKeeper("localhost:5181", 1000 * 60, zkInitWatcher);
            myWatch.ZkDemo = this;
        }

        public bool WaitConnected(TimeSpan timeout)
        {
            var continueWait = false;
            while (this.state != Event.KeeperState.SyncConnected)
            {
                continueWait = notifyEvent.WaitOne(timeout);
                if (!continueWait)
                {
                    return false;
                }
            }
            return true;
        }

        private MyWatch myWatch = new MyWatch();

        private object lockObject = new object();
        public bool HasAcquireLock { get; set; }

        public void WatchPath(String path)
        {
            //var myWatcher = new MyWatch();
            myWatch.WatchedPath = path;

            var spath = this.ZK.createAsync("/zk/node-", Encoding.UTF8.GetBytes("data"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL).Result;
            myWatch.EpemeralPath = spath;

            var curId = GetIdFromPath(spath);
            myWatch.NodeId = curId;

            //Log.Debug("WatchPath {0}, Tid: {1}", path, Thread.CurrentThread.ManagedThreadId);

            //var zkState = ZK.getState();
            //Log.Debug("CurrentZkState {@state}", zkState);

            var reuslt = ZK.getChildrenAsync("/zk", myWatch).GetAwaiter().GetResult();

            //Log.Debug("WatchPath Result {@r}", reuslt.Children);


            Log.Debug("=========>Begin CurrentId {@id}", curId);

            var ids = new List<long>();

            foreach (var item in reuslt.Children)
            {
                //Log.Debug("Children: {@c}", item);
                ids.Add(GetIdFromPath(item));
            }

            if (ids.Count > 0 && ids.Min() == curId)
            {
                lock (this)
                {
                    //Log.Debug("CurrentId {@id} Acquire Lock Directly Invoke Callback.", curId);

                    this.HasAcquireLock = true;
                    this.FireAcquireLock(curId).Wait();
                    //Log.Debug("CurrentId {@id} Acquire Lock Directly AND Release Lock Directly.", curId);
                }
            }
        }

        public long GetIdFromPath(string path)
        {
            var m = Regex.Match(path, "(\\d+)");
            if (m.Success)
            {
                return long.Parse(m.Groups[0].Value);
            }
            return 0L;
        }

        public async static Task TestWatch(Watcher myWatcher)
        {
            var pwd = Encoding.UTF8.GetBytes("123");
            var zooKeeper = new ZooKeeper("localhost:5181", 1000 * 60, null);

            var state = zooKeeper.getState();
            Console.WriteLine(state);

            while (zooKeeper.getState() != org.apache.zookeeper.ZooKeeper.States.CONNECTED)
            {
                Thread.Sleep(50);
            }
            state = zooKeeper.getState();
            Console.WriteLine(state);

            var stat = await zooKeeper.getChildrenAsync("/zk", myWatcher);
            Log.Logger.Debug("state: {@stat}", stat);
        }

        public async static Task TestConnect()
        {
            Console.WriteLine("connect zookeeper");
            org.apache.zookeeper.ZooKeeper zooKeeper = new org.apache.zookeeper.ZooKeeper("localhost:5181", 1000, null);

            var state = zooKeeper.getState();
            Console.WriteLine(state);

            while (zooKeeper.getState() != org.apache.zookeeper.ZooKeeper.States.CONNECTED)
            {
                Thread.Sleep(50);
            }
            state = zooKeeper.getState();
            Console.WriteLine(state);

            var parent = "/zk";
            var prefix = "lock/list";


            var path = $"{parent}/{prefix}";

            //var nodePath = await zooKeeper.createAsync(path, Encoding.UTF8.GetBytes("lock list"), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            //    CreateMode.PERSISTENT);

            //Console.WriteLine($"nodePath: {nodePath}");

            var chilren = await zooKeeper.getChildrenAsync(path, new MyWatch());

            Console.ReadLine();


            //await zooKeeper.deleteAsync("/zk/node1");
            //Console.WriteLine("Delete Done");


            //var node = await zooKeeper.createAsync(path, Encoding.UTF8.GetBytes("tmp data"),
            //    ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode:CreateMode.EPHEMERAL_SEQUENTIAL);

            //Console.WriteLine("nodeid: {0}", node);

            //node = await zooKeeper.createAsync(path, Encoding.UTF8.GetBytes("tmp data"),
            //    ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode: CreateMode.EPHEMERAL_SEQUENTIAL);

            //Console.WriteLine("nodeid: {0}", node);

            //var info = await zooKeeper.getChildrenAsync("/zk", null);

            //Console.WriteLine(info);

            //var data = await zooKeeper.getDataAsync("/zk");
            //var dataOfString = Encoding.UTF8.GetString(data.Data);
            //Console.WriteLine($"/zk data {dataOfString}");


            await zooKeeper.closeAsync();
        }
    }
}
